package com.hz.cheetah.client;

import com.hz.cheetah.common.Constants;
import com.hz.cheetah.common.bean.RpcRequest;
import com.hz.cheetah.common.bean.RpcResponse;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by yangchuan on 16/7/4.
 */
public class DefaultFuture implements ResponseFuture {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultFuture.class);

    private static final Map<String, DefaultFuture> FUTURES   = new ConcurrentHashMap<String, DefaultFuture>();

    private final Lock lock = new ReentrantLock();

    private final Condition done = lock.newCondition();

    private volatile RpcResponse response;

    private String requstId;

    private final int                             timeout;
    private final long                            start = System.currentTimeMillis();

    private CChannel                              channel;


    public DefaultFuture(String id,CChannel cChannel,int timeout){
        this.requstId = id;
        this.channel = cChannel;
        FUTURES.put(id,this);
        this.timeout = timeout > 0 ? timeout : Constants.DEFAULT_TIMEOUT;
    }

    public DefaultFuture(String id,CChannel cChannel){
        this(id,cChannel,Constants.DEFAULT_TIMEOUT);
    }

    private int getTimeout() {
        return timeout;
    }

    private long getStartTimestamp() {
        return start;
    }

    public String getRequestId(){
        return this.requstId;
    }

    public CChannel getChannel(){
        return this.channel;
    }


    @Override
    public RpcResponse get() {
        return get(timeout);
    }

    @Override
    public RpcResponse get(int timeoutInMillis) {
        if (! isDone()) {
            long start = System.currentTimeMillis();
            lock.lock();
            try {
                while (! isDone()) {
                    /**
                     * 使用timeout没有关系,在socket 收到消息后,就会马上done.signal() 让
                     * await 停止等到,往下执行。
                     */
                    done.await(timeout, TimeUnit.MILLISECONDS);
                    if (isDone() || System.currentTimeMillis() - start > timeoutInMillis) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
            if (! isDone()) {
                throw new RPCTimeOutException();
            }
        }



        return returnFromResponse();
    }

    private RpcResponse returnFromResponse()  {
        if(response.hasException()){
            throw new RuntimeException(response.getException());
        }
        //TODO response

        return response;
    }


    public static void  received(RpcResponse response){
        if(response.getType() == 1){

        }
        DefaultFuture future = DefaultFuture.remove(response.getRequestId());
        if (future != null) {
            future.doReceived(response);
        } else {
            CChannel channel = future.getChannel();
            LOGGER.warn("The timeout response finally returned at "
                    + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
                    + ", response " + response
                    + (channel == null ? "" : ", channel: " + channel.remoteAddress()
                    + " -> " + channel.localAddress()));
            channel = null;
        }
    }

    public void doReceived(RpcResponse res) {
        lock.lock();
        try {
            response = res;
            if (done != null) {
                done.signal();
            }
        } finally {
            lock.unlock();
        }
    }

    public static DefaultFuture remove(String id){
        return FUTURES.remove(id);
    }

    @Override
    public boolean isDone() {
        return response != null;
    }


    private static class RemotingInvocationTimeoutScan implements Runnable {

        public void run() {
            while (true) {
                try {
                    for (DefaultFuture future : FUTURES.values()) {
                        if (future == null || future.isDone()) {
                            continue;
                        }
                        if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                            RpcResponse response = new RpcResponse();
                            response.setRequestId(future.getRequestId());
                            response.setException(new RPCTimeOutException());
                            // handle response.
                            DefaultFuture.received(response);
                        }
                    }
                    Thread.sleep(30);
                } catch (Throwable e) {
                    LOGGER.error("Exception when scan the timeout invocation of remoting.", e);
                }
            }
        }
    }

    static {
        Thread th = new Thread(new RemotingInvocationTimeoutScan(), "ResponseTimeoutScanTimer");
        th.setDaemon(true);
        th.start();
    }
}
